package com.kd.crawler.common.jms;

import javax.jms.Connection;
import javax.jms.Destination;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageConsumer;
import javax.jms.Session;
import javax.jms.Topic;


/**
 * 消息的消费者（接受者）
 * 
 * @author liang
 *
 */
public class ActiveMQConsumer {
	
	public static void recevieTopicMessage(String topicName,String clientID,final ReceiveListener listener) {
		// 连接
		Connection connection = null;
		// 会话 接受或者发送消息的线程
		Session session=null;
		try {
			connection = MQPooledConnectionFactory.getConnection();
			if(clientID!=null&&!clientID.trim().equals("")){
				connection.setClientID(clientID);
			}
			// 启动连接
			connection.start();
			// 创建session
			session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
			Topic topic = session.createTopic(topicName);
			String subscriberName= clientID==null?topicName+"_":topicName+"_"+clientID;
			// 消息生产者
			MessageConsumer messageConsumer =session.createDurableSubscriber(topic,subscriberName);		
			// 发送消息
			while (true) {
				if(listener!=null){
					listener.onReceive(messageConsumer.receive(10000));
				}
				//session.commit();
			}
			
			
		} catch (Exception e) {
			e.printStackTrace();
		} finally {
			if(session!=null){
				try {
					session.close();
				} catch (JMSException e) {
					e.printStackTrace();
				}
			}			
			if (connection != null) {
				try {
					connection.close();
				} catch (JMSException e) {
					e.printStackTrace();
				}
			}
		}

	}
	

	public static void recevieQueueMessage(String queueName,ReceiveListener listener) {
		// 连接
		Connection connection = null;
		// 会话 接受或者发送消息的线程
		Session session=null;
		MessageConsumer messageConsumer=null;
		while(true){
			try {
				connection = MQPooledConnectionFactory.getConnection();
				// 启动连接
				connection.start();
				// 创建session
				session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
				Destination destination = session.createQueue(queueName);
				// 消息生产者
				messageConsumer = session.createConsumer(destination);
				// 发送消息
				while (true) {
					if(listener!=null){
						listener.onReceive(messageConsumer.receive());	
						//session.commit();
					}
				}
			} catch (Exception e) {
				e.printStackTrace();
			} finally {
				if(messageConsumer!=null){
					try {
						messageConsumer.close();
					} catch (JMSException e) {
						e.printStackTrace();
					}
				}
				if(session!=null){
					try {
						session.close();
					} catch (JMSException e) {
						e.printStackTrace();
					}
				}
				if (connection != null) {
					try {
						connection.close();
					} catch (JMSException e) {
						e.printStackTrace();
					}
				}
			}
		}

	}


	public interface ReceiveListener {
		public void onReceive(Message message);
	}

}